package com.flx.dis.registry.zk.service.impl;

import com.flx.dis.registry.zk.server.ZkServer;
import com.flx.dis.registry.zk.service.MasterSelectService;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @Author Fenglixiong
 * @Create 2022/4/21 22:44
 * @Description
 **/
@Slf4j
@Service
public class MasterSelectServiceImpl implements MasterSelectService {

    @Autowired
    private ZkServer zkServer;

    //跟节点
    private final static String rootPath = "/server-manager";
    //临时序号节点
    private static String serverPath = rootPath+"/server";
    //当前路径节点
    private String currentPath;
    //是否是master节点
    private volatile boolean master = false;

    public void startMasterSelect()throws Exception{

        buildRoot();
        createServerNode();

    }

    /**
     * 创建持久化跟节点
     * @throws Exception
     */
    public void buildRoot()throws Exception{
        if(!zkServer.exists(rootPath)) {
            zkServer.createPersistent(rootPath,"root");
            log.info("成功创建持久跟节点：path = {}",rootPath);
        }
    }

    /**
     * 创建当前服务临时节点
     * @throws Exception
     */
    public void createServerNode()throws Exception{
        currentPath = zkServer.createEphemeralSequential(serverPath,"slave");
        log.info("成功创建服务临时节点：path = {}",serverPath);
        initMaster();
        initListener();
    }

    /**
     * 初始化master节点
     * @throws Exception
     */
    public void initMaster()throws Exception{
        boolean existMaster =  zkServer.getChildren(rootPath)
                .stream()
                .map(e->rootPath+"/"+e)
                .map(e-> {
                    try {
                        return zkServer.getData(e);
                    } catch (Exception ex) {
                        ex.printStackTrace();
                        return null;
                    }
                })
                .anyMatch(e-> Objects.equals(e,"master"));
        if(!existMaster){
            zkServer.setData(currentPath,"master");
            master = true;
            log.info("当前节点当选master...");
        }
    }

    /**
     * 监听子节点变化
     * @throws Exception
     */
    public void initListener()throws Exception{
        zkServer.subscribeChildChange(rootPath,(parentPath,children)->{
            doSelection();
        });
    }

    /**
     * 如果有节点挂掉
     * 选举master操作
     * @throws Exception
     */
    public void doSelection()throws Exception{
        Map<String, String> childData = zkServer.getChildren(rootPath)
                .stream()
                .map(e -> rootPath + "/" + e)
                .collect(Collectors.toMap(e -> e, e -> {
                    try {
                        return zkServer.getData(e);
                    } catch (Exception ex) {
                        ex.printStackTrace();
                        return null;
                    }
                }));
        //如果子节点有master节点，就不用重新选举master节点
        if(childData.containsValue("master")){
            return;
        }
        childData.keySet().stream().sorted().findFirst().ifPresent(e->{
            if(e.equals(currentPath)){
                try {
                    zkServer.setData(currentPath,"master");
                    master = true;
                    log.info("当前当选master！");
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        });
    }

}
